package com.shujia.transformaction

import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

object Demo3KeyByOperator {
  def main(args: Array[String]): Unit = {


    //创建flink  程序的入口  相当于sparkContext
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(2)

    val ds = env.readTextFile("data/students.txt")

    val studentDS = ds.map(line => {
      val split = line.split(",")
      val id = split(0)
      val name = split(1)
      val age = split(2).toInt
      val gender = split(3)
      val clazz = split(4)
      Student(id, name, age, gender, clazz, 1)
    })

    /**
      * keyBy  将相同的key发送到同一个task中
      *
      */


    //统计班级人数
    //studentDS
    //      .keyBy(_.clazz)//指定属性
    //      .keyBy("clazz")//指定字段名
    //.keyBy(4) //指定下标
    //.sum("i")
    //.print()

    studentDS.keyBy(_.clazz).print()

    env.execute("KeyBy")


  }

  case class Student(id: String, name: String, age: Int, gender: String, clazz: String, i: Int)

}
